package edu.zju.gis.dbfg.model.statistic;
import edu.zju.gis.dbfg.common.exception.ModelFailedException;
import edu.zju.gis.dbfg.model.PyramidConfig;
import edu.zju.gis.dbfg.model.datasource.DataReaderRDD;
import edu.zju.gis.dbfg.model.util.CrsPick;
import edu.zju.gis.dbfg.model.util.GeoItem;
import edu.zju.gis.dbfg.model.util.TileID;
import edu.zju.gis.dbfg.model.util.WriteToFile;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.*;
import org.geotools.referencing.CRS;
import org.locationtech.jts.geom.Geometry;
import org.opengis.referencing.crs.CoordinateReferenceSystem;
import org.opengis.referencing.operation.MathTransform;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.Tuple2;

import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.util.*;

import static edu.zju.gis.dbfg.model.base.Constant.*;
public class LUXZQStatistic {

    private final static Logger logger = LoggerFactory.getLogger(LUXZQStatistic.class);
    public static void main(String[] args) throws Exception {
        logger.info("Setup Spark Context");
        SparkConf conf = new SparkConf();
        conf.setMaster("local[4]");
        conf.setAppName("lalala");
        JavaSparkContext jsc = new JavaSparkContext(conf);

        logger.info("Setup arguments");
        LUXZQStatisticArgs mArgs = LUXZQStatisticArgs.initArgs(args);
        if (mArgs == null) {
            String msg = "init args failed, exit";
            logger.error(msg);
            throw new ModelFailedException(LUXZQStatistic.class, "main()", msg, args);
        }

        CoordinateReferenceSystem crs2 = CRS.parseWKT(CrsPick.pickByName(mArgs.getCrs2()));
        CoordinateReferenceSystem targetCRS = DEFAULT_CRS();
        PyramidConfig targetPyramidConfig = DEFAULT_PYRAMID_CONFIG();//gewang

        MathTransform transform2 = null;
        if (!crs2.equals(targetCRS)) {
            transform2 = CRS.findMathTransform(crs2, targetCRS);
        }

        String input2 = mArgs.getInput2();
        String type2 = mArgs.getType2();
        String zldwdm2 = mArgs.getZldwdmIndex2();
        String dlbm2 = mArgs.getDlbmIndex2();
        String area2 = mArgs.getAreaIndex2();
        String jarea2 = mArgs.getJareaIndex2();
        String kcdlbm2 = mArgs.getKcdlbmIndex2();
        String xzq2 = mArgs.getXzq();


        Map<String, Object> attrs2 = new HashMap<>();
        attrs2.put("DLBM", dlbm2);
        attrs2.put("ZLDWDM", zldwdm2);
        attrs2.put("TBMJ", area2);
        attrs2.put("TBDLMJ", jarea2);
        attrs2.put("KCDLBM", kcdlbm2);
        JavaRDD<GeoItem> gi2Pair = null;

        if (type2.equals("ES")) {
            gi2Pair = DataReaderRDD.readToRDD2(type2, jsc, input2, attrs2, null, null, transform2);
        } else {
            gi2Pair = DataReaderRDD.readToRDD(type2, jsc, input2, "1", attrs2, transform2);
        }

        JavaPairRDD<String,GeoItem> tempRDD=gi2Pair.mapToPair(new PairFunction<GeoItem, String, GeoItem>() {
            @Override
            public Tuple2<String, GeoItem> call(GeoItem gi) throws Exception {

                String dlbm = String.valueOf(gi.getProp().get("DLBM")) .replace("\"","").replace("\40","");
                String zldwdm = String.valueOf(gi.getProp().get("ZLDWDM")).replace("\"","").replace("\40","").substring(0,9);
                //Double tbdlmj=Double.valueOf((String) gi.getProp().get("TBDLMJ"));
                Double tbdlmj=Double.valueOf(gi.getProp().get("TBDLMJ").toString());
                String kcdlbm = String.valueOf(gi.getProp().get("KCDLBM")).replace("\"","").replace("\40","");
                Double tbmj = Double.valueOf(gi.getProp().get("TBMJ").toString());

                if(zldwdm.startsWith(xzq2)){

                    String id = zldwdm + "##" + dlbm;
                    GeoItem result = new GeoItem();
                    result.setFid(id);
                    Map<String, Object> props = new HashMap<>();
                    props.put("ZLDWDM", zldwdm);
                    props.put("DLBM", dlbm);
                    props.put("KCDLBM",kcdlbm);
                    props.put("TBMJ",tbmj);
                    props.put("AREA", tbdlmj);
                    result.setProp(props);

                    return new Tuple2<>(id, result);
                }

                return new Tuple2<>("EMPTY", null);
            }
        }).filter(new Function<Tuple2<String, GeoItem>, Boolean>() {
            @Override
            public Boolean call(Tuple2<String, GeoItem> input) throws Exception {
                return !input._1.equals("EMPTY");
            }
        });

        JavaPairRDD<String, GeoItem> kctempRDD = tempRDD.mapToPair(new PairFunction<Tuple2<String, GeoItem>, String, GeoItem>() {
            @Override
            public Tuple2<String, GeoItem> call(Tuple2<String, GeoItem> input) throws Exception {
                String kcdlbm = String.valueOf(input._2.getProp().get("KCDLBM"));
                if(!kcdlbm.equals(null) && kcdlbm.length()>3){
                    String zldwdm = String.valueOf(input._2.getProp().get("ZLDWDM"));
                    Double kcmj = Double.valueOf(String.valueOf(input._2.getProp().get("TBMJ")))-Double.valueOf(String.valueOf(input._2.getProp().get("AREA").toString()));
                    String id = zldwdm + "##" + kcdlbm;
                    GeoItem result = new GeoItem();
                    result.setFid(id);

                    Map<String, Object> props = new HashMap<>();
                    props.put("DLBM", kcdlbm);
                    props.put("ZLDWDM", zldwdm);
                    props.put("AREA", kcmj);
                    result.setProp(props);

                    return new Tuple2<>(id, result);
                }
                return new Tuple2<>("EMPTY", null);
            }
        }).filter(new Function<Tuple2<String, GeoItem>, Boolean>() {
            @Override
            public Boolean call(Tuple2<String, GeoItem> input) throws Exception {
                return !input._1.equals("EMPTY");
            }
        });
        tempRDD.cache();
        kctempRDD.cache();

        JavaPairRDD<String, GeoItem> alltempRDD=tempRDD.union(kctempRDD);

        JavaPairRDD<String, GeoItem> resultRDD = alltempRDD.reduceByKey(new Function2<GeoItem, GeoItem, GeoItem>() {
            @Override
            public GeoItem call(GeoItem gi1, GeoItem gi2) throws Exception {
                Double area = (Double) gi1.getProp().get("AREA") + (Double) gi2.getProp().get("AREA");
                GeoItem result = new GeoItem();
                String id=gi1.getFid();
                result.setFid(id);
                Map<String, Object> props = new HashMap<>();
                props.put("AREA", area);
                result.setProp(props);
                return result;
            }
        }).sortByKey();

        if(xzq2.length()<6){
            JavaPairRDD<String,GeoItem> xianRDD=alltempRDD.mapToPair(new PairFunction<Tuple2<String, GeoItem>, String, GeoItem>() {
                @Override
                public Tuple2<String, GeoItem> call(Tuple2<String, GeoItem> input) throws Exception {
                    String zldwdm = String.valueOf(input._2.getProp().get("ZLDWDM")).substring(0,6);
                    String dlbm = String.valueOf(input._2.getProp().get("DLBM"));
                    Double area = Double.valueOf(String.valueOf(input._2.getProp().get("AREA")));
                    String id = zldwdm + "##" + dlbm;
                    GeoItem result = new GeoItem();
                    result.setFid(id);
                    Map<String, Object> props = new HashMap<>();
                    props.put("DLBM", dlbm);
                    props.put("ZLDWDM", zldwdm);
                    props.put("AREA", area);
                    result.setProp(props);

                    return new Tuple2<>(id, result);

                }
            }).filter(new Function<Tuple2<String, GeoItem>, Boolean>() {
                @Override
                public Boolean call(Tuple2<String, GeoItem> input) throws Exception {
                    return !input._1.equals("EMPTY");
                }
            }).reduceByKey(new Function2<GeoItem, GeoItem, GeoItem>() {
                @Override
                public GeoItem call(GeoItem gi1, GeoItem gi2) throws Exception {
                    Double area = (Double) gi1.getProp().get("AREA") + (Double) gi2.getProp().get("AREA");
                    GeoItem result = new GeoItem();
                    String id=gi1.getFid();
                    result.setFid(id);
                    Map<String, Object> props = new HashMap<>();
                    props.put("AREA", area);
                    result.setProp(props);
                    return result;

                }
            });
            resultRDD=resultRDD.union(xianRDD).sortByKey();

        }

        if(xzq2.length()<4){
            JavaPairRDD<String,GeoItem> shiRDD=alltempRDD.mapToPair(new PairFunction<Tuple2<String, GeoItem>, String, GeoItem>() {
                @Override
                public Tuple2<String, GeoItem> call(Tuple2<String, GeoItem> input) throws Exception {
                    String zldwdm = String.valueOf(input._2.getProp().get("ZLDWDM")).substring(0,4);
                    String dlbm = String.valueOf(input._2.getProp().get("DLBM"));
                    Double area = Double.valueOf(String.valueOf(input._2.getProp().get("AREA")));
                    String id = zldwdm + "##" + dlbm;
                    GeoItem result = new GeoItem();
                    result.setFid(id);
                    Map<String, Object> props = new HashMap<>();
                    props.put("DLBM", dlbm);
                    props.put("ZLDWDM", zldwdm);
                    props.put("AREA", area);
                    result.setProp(props);

                    return new Tuple2<>(id, result);

                }
            }).filter(new Function<Tuple2<String, GeoItem>, Boolean>() {
                @Override
                public Boolean call(Tuple2<String, GeoItem> input) throws Exception {
                    return !input._1.equals("EMPTY");
                }
            }).reduceByKey(new Function2<GeoItem, GeoItem, GeoItem>() {
                @Override
                public GeoItem call(GeoItem gi1, GeoItem gi2) throws Exception {
                    Double area = (Double) gi1.getProp().get("AREA") + (Double) gi2.getProp().get("AREA");
                    GeoItem result = new GeoItem();
                    String id=gi1.getFid();
                    result.setFid(id);
                    Map<String, Object> props = new HashMap<>();
                    props.put("AREA", area);
                    result.setProp(props);
                    return result;

                }
            });
            resultRDD=resultRDD.union(shiRDD).sortByKey();
        }

        if(xzq2.length()<2){
            JavaPairRDD<String,GeoItem> shengRDD=alltempRDD.mapToPair(new PairFunction<Tuple2<String, GeoItem>, String, GeoItem>() {
                @Override
                public Tuple2<String, GeoItem> call(Tuple2<String, GeoItem> input) throws Exception {
                    String zldwdm = String.valueOf(input._2.getProp().get("ZLDWDM")).substring(0,2);
                    String dlbm = String.valueOf(input._2.getProp().get("DLBM"));
                    Double area = Double.valueOf(String.valueOf(input._2.getProp().get("AREA")));
                    String id = zldwdm + "##" + dlbm;
                    GeoItem result = new GeoItem();
                    result.setFid(id);
                    Map<String, Object> props = new HashMap<>();
                    props.put("DLBM", dlbm);
                    props.put("ZLDWDM", zldwdm);
                    props.put("AREA", area);
                    result.setProp(props);

                    return new Tuple2<>(id, result);

                }
            }).filter(new Function<Tuple2<String, GeoItem>, Boolean>() {
                @Override
                public Boolean call(Tuple2<String, GeoItem> input) throws Exception {
                    return !input._1.equals("EMPTY");
                }
            }).reduceByKey(new Function2<GeoItem, GeoItem, GeoItem>() {
                @Override
                public GeoItem call(GeoItem gi1, GeoItem gi2) throws Exception {
                    Double area = (Double) gi1.getProp().get("AREA") + (Double) gi2.getProp().get("AREA");
                    GeoItem result = new GeoItem();
                    String id=gi1.getFid();
                    result.setFid(id);
                    Map<String, Object> props = new HashMap<>();
                    props.put("AREA", area);
                    result.setProp(props);
                    return result;

                }
            }).sortByKey();
            resultRDD=resultRDD.union(shengRDD).sortByKey();
        }

        Map<String, GeoItem> result = resultRDD.collectAsMap();
        File file = new File(mArgs.getOutput());
        if (file.exists()) {
            file.delete();
        }
        FileWriter writer = new FileWriter(file);
        WriteToFile.writeToFile(writer,result);
        writer.close();
    }

}
